package com.gzyj.flink.gps.map.prohibitdriving;

import cn.hutool.core.lang.Pair;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson.support.spring.FastJsonRedisSerializer;
import com.gzyj.flink.driver.Driverlogininfo;
import com.gzyj.flink.dto.alarm.Kafka_VehicleRemindContent;
import com.gzyj.flink.dto.alarm.ProhibitDrivingData;
import com.gzyj.flink.dto.alarm.RemindContent;
import com.gzyj.flink.gps.GPSPosition;
import com.gzyj.flink.gps.config.RedisKeyPrefixConfig;
import com.gzyj.flink.gps.factory.MySQLMybatisSessionFactory;
import com.gzyj.flink.gps.factory.RedisSessionFactory;
import com.gzyj.flink.gps.map.fatiguedriving.FatigueDrivingResult;
import com.gzyj.flink.gps.mapper.mysql.MonYanzhongweiguiSetMapper;
import com.gzyj.flink.gps.mapper.mysql.MonYejianSetMapper;
import com.gzyj.flink.gps.mapper.mysql.TsAlarmRuleMapper;
import com.gzyj.flink.gps.service.NormalAlarmService;
import com.gzyj.flink.gps.utils.TimeUtils;
import com.gzyj.flink.pojo.mysql.set.MonYanzhongweiguiSet;
import com.gzyj.flink.pojo.mysql.set.MonYejianSet;
import com.gzyj.flink.pojo.mysql.set.TsAlarmRule;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class ProhibitDrivingDataProcess extends RichMapFunction<GPSPosition, FatigueDrivingResult> {
    LettuceConnectionFactory redisFactory;

    SqlSessionFactory mysql;

    NormalAlarmService normalAlarmService;

    private RedisTemplate<String, Object> getRedisTemplate(LettuceConnectionFactory redisSessionFactory) {
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisSessionFactory);
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new FastJsonRedisSerializer<>(Object.class));
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(new FastJsonRedisSerializer<>(Object.class));
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }


    @Override
    public void open(Configuration parameters) throws Exception {
        redisFactory = RedisSessionFactory.getRedisFactory();
        mysql = MySQLMybatisSessionFactory.getSqlSessionFactory();
        normalAlarmService = new NormalAlarmService();
    }


    private MonYejianSet getVehicleForFatigueParamsDic(String vehId, MonYejianSetMapper monYejianSetMapper) {
        return monYejianSetMapper.selectByVehID(vehId);
    }

    public Pair<Boolean, MonYejianSet> GetRuleSetting(GPSPosition vehicle, MonYejianSetMapper monYejianSetMapper, TsAlarmRuleMapper tsAlarmRuleMapper) {


        MonYejianSet setting = getVehicleForFatigueParamsDic(vehicle.getVehId(), monYejianSetMapper);
        if (setting != null) {
            return new Pair<>(true, setting);
        } else {


            TsAlarmRule rule = tsAlarmRuleMapper.selectByRuleTypeYejian(vehicle.getShiyongxingzhi());
            if (rule == null || rule.getRuleFlag() == 0) {
                return new Pair<>(false, null);
            } else {
                setting = getVehicleForFatigueParamsDic(rule.getRuleNo(), monYejianSetMapper);
                if (setting != null) {
                    return new Pair<>(true, setting);
                } else {
                    return new Pair<>(false, null);
                }
            }
        }
    }

    private RemindContent GenerateVehicleRemind(ProhibitDrivingData data, MonYejianSet setting, GPSPosition gps, String msgType) {
        try {
            if (msgType.equalsIgnoreCase("预警短信")) {
                RemindContent content = new RemindContent() {
                    {
                        VehicleId = gps.getVehId();
                        Ipaddress = gps.getTerminalPhoneNo();
                        VehicleNo = gps.getCheliangpaizhao();
                        PlateColor = gps.getChepaiyanse();
                        RemindNote = setting.getTxnr();
                        RemindType = 0x01;
                        RemindTime = new Date();
                        MessageType = "夜间禁行报警";
                        AlarmNote = "夜间禁行报警";
                        AlarmGuid = data.SingleGuid;
                        GpsTime = data.GpsTime;
                        LastSpeed = gps.getSpeedExt();
                    }

                };
                return content;
            } else if (msgType.equalsIgnoreCase("报警短信")) {
                RemindContent content = new RemindContent() {
                    {
                        VehicleId = gps.getVehId();
                        Ipaddress = gps.getTerminalPhoneNo();
                        VehicleNo = gps.getCheliangpaizhao();
                        PlateColor = gps.getChepaiyanse();
                        RemindNote = setting.getBjnr();
                        RemindType = 0x02;
                        RemindTime = new Date();
                        MessageType = "夜间禁行报警";
                        AlarmNote = "夜间禁行报警";
                        AlarmGuid = data.SingleGuid;
                        GpsTime = data.GpsTime;
                        LastSpeed = gps.getSpeedExt();
                    }

                };
                return content;
            } else {
                return null;
            }
        } catch (Exception ex) {
            ex.printStackTrace();
            return null;
        }
    }

    private Kafka_VehicleRemindContent DisposeMessage(ProhibitDrivingData data, MonYejianSet mySetting, GPSPosition gps, String msgType) {
        if (msgType.equalsIgnoreCase("预警短信")) {
            if (mySetting.getTxcs() > 0 && StringUtils.isNotEmpty(mySetting.getTxnr())) {
                long ts = TimeUtils.getMinusSecond(new Date(), data.RemindContentSendTime);
                if (ts >= mySetting.getTxjg() && data.RemindContentCount < mySetting.getTxcs()) {
                    data.IsRemindContent = true;
                    data.RemindContentDate = TimeUtils.toDate(TimeUtils.getToday());
                    data.RemindContentCount++;
                    data.RemindContentSendTime = new Date();
                    RemindContent remindContent = GenerateVehicleRemind(data, mySetting, gps, msgType);
                    return normalAlarmService.SendRemindToKafka(remindContent);
                }
            }
        } else if (msgType.equalsIgnoreCase("报警短信")) {
            if (mySetting.getBjcs() > 0 && StringUtils.isNotEmpty(mySetting.getBjnr())) {
                long ts = TimeUtils.getMinusSecond(new Date(), data.AlarmContentSendTime);
                if (ts >= mySetting.getBjjg() && data.AlarmContentCount < mySetting.getBjcs()) {
                    data.AlarmContentCount++;
                    data.AlarmContentSendTime = new Date();
                    RemindContent message = GenerateVehicleRemind(data, mySetting, gps, msgType);
                    return normalAlarmService.SendRemindToKafka(message);
                }
            }
        }
        return null;
    }

    public Driverlogininfo TryGetData_DLD(String vehId, RedisTemplate<String, Object> clusterRedisTemplate) {
        Object obj = clusterRedisTemplate.opsForValue().get(RedisKeyPrefixConfig.RedisKey_DriverSignOn + vehId);
        if (obj == null) {
            return new Driverlogininfo();
        } else {
            return JSON.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), Driverlogininfo.class);
        }

    }

    private MonYanzhongweiguiSet getVehicleForSeverityAlarmParamsDic(String vehiId, MonYanzhongweiguiSetMapper monYanzhongweiguiSetMapper) {
        return monYanzhongweiguiSetMapper.selectByVehId(vehiId);
    }

    public Pair<Boolean, MonYanzhongweiguiSet> GetSeverityRuleSetting(GPSPosition vehicle, MonYanzhongweiguiSetMapper monYanzhongweiguiSetMapper, TsAlarmRuleMapper tsAlarmRuleMapper) {
        MonYanzhongweiguiSet setting = getVehicleForSeverityAlarmParamsDic(vehicle.getVehId(), monYanzhongweiguiSetMapper);
        if (setting != null) {
            return new Pair<>(true, setting);
        } else {
            List<TsAlarmRule> list = tsAlarmRuleMapper.selectByRuleTypeYanzhong(vehicle.getShiyongxingzhi());

            TsAlarmRule rule = list.size() > 0 ? list.get(0) : null;
            if (rule == null || rule.getRuleFlag() == 0) {
                return new Pair<>(false, null);
            } else {
                setting = getVehicleForSeverityAlarmParamsDic(rule.getRuleNo(), monYanzhongweiguiSetMapper);
                if (setting != null) {
                    return new Pair<>(true, setting);
                } else {
                    return new Pair<>(false, null);
                }
            }
        }
    }

    @Override
    public FatigueDrivingResult map(GPSPosition gpsPosition) throws Exception {
        RedisTemplate<String, Object> clusterRedisTemplate = getRedisTemplate(redisFactory);
        SqlSession sqlSession = mysql.openSession(true);
        FatigueDrivingResult prohibitDrivingResult = new FatigueDrivingResult();


        try {
            MonYejianSetMapper monYejianSetMapper = sqlSession.getMapper(MonYejianSetMapper.class);
            TsAlarmRuleMapper tsAlarmRuleMapper = sqlSession.getMapper(TsAlarmRuleMapper.class);
            MonYanzhongweiguiSetMapper monYanzhongweiguiSetMapper = sqlSession.getMapper(MonYanzhongweiguiSetMapper.class);


            Driverlogininfo driverLoginData = TryGetData_DLD(gpsPosition.getVehId(), clusterRedisTemplate);
            ProhibitDrivingData prohibitDrivingData = TryGetData_PDD(gpsPosition.getVehId(), clusterRedisTemplate);
            if (prohibitDrivingData == null) return null;
            Pair<Boolean, Integer> p = TimeUtils.CheckData(prohibitDrivingData.getGpsTime(), gpsPosition.getTIME());
            int i = p.getValue();

            if (!p.getKey()) {
                return null;
            } else {
                prohibitDrivingData.LocalTime = new Date();
                prohibitDrivingData.GpsTime = gpsPosition.getGPSTimeExt();
            }

            Pair<Boolean, MonYejianSet> p1 = GetRuleSetting(gpsPosition, monYejianSetMapper, tsAlarmRuleMapper);
            MonYejianSet mySetting = p1.getValue();

            if (p1.getKey()) {
                //如果车辆配置了规则
                if (mySetting.getEnable() == 1) {


                    // 夜间开始时间前发送预警信息，这里预警得GUID是关联不上报警得
                    prohibitDrivingData.RemindReset();
                    long tsGpsTime = TimeUtils.getHour(gpsPosition.getTIME());
                    long tsJykssj = TimeUtils.getHour(mySetting.getYjkssj());
                    boolean isRemind = tsGpsTime >= tsJykssj - mySetting.getYjtqsj() * 60 && tsGpsTime <= tsJykssj;
                    if (isRemind && gpsPosition.getSpeedExt() > mySetting.getTcpdsd()) {
                        prohibitDrivingResult.setKafkaVehicleRemindContent(DisposeMessage(prohibitDrivingData, mySetting, gpsPosition, "报警短信"));
                    }
                    boolean inTimeRange = TimeUtils.InTimeRange(gpsPosition.getGPSTimeExt(), mySetting.getYjkssj(), mySetting.getYjjssj());
                    if (inTimeRange) {
                        if (gpsPosition.getSpeedExt() > mySetting.getTcpdsd()) {

                            if (!prohibitDrivingData.IsCalc) {
                                prohibitDrivingData.ResetData();
                                prohibitDrivingData.StartCalc(gpsPosition);
                            }
                            prohibitDrivingData.WriteCalc(gpsPosition);
                            if (prohibitDrivingData.CalcConditionTimeSpan() > mySetting.getTccxsj()*60) {
                                //满足报警条件
                                if (!prohibitDrivingData.IsAlarm) {
                                    prohibitDrivingData.StartAlarm(gpsPosition);
                                    //从没报警切换到报警，发送开始报警
                                    prohibitDrivingResult.setKafkaVehicleNormalAlarm(normalAlarmService.SendAlarm(prohibitDrivingData, gpsPosition, true, driverLoginData));
                                }
                                //其他的时候记录报警的过程数据
                                prohibitDrivingData.WriteAlarm(gpsPosition);
                                prohibitDrivingResult.setKafkaVehicleRemindContent(DisposeMessage(prohibitDrivingData, mySetting, gpsPosition, "报警短信"));
                                // 改变报警级别
                                Pair<Boolean, MonYanzhongweiguiSet> p2 = GetSeverityRuleSetting(gpsPosition, monYanzhongweiguiSetMapper, tsAlarmRuleMapper);
                                MonYanzhongweiguiSet severityAlarmSetting = p2.getValue();
                                if (p2.getKey()) {
                                    prohibitDrivingData.AlarmSetting = 1;
                                    // 通用规则
                                    double interval = TimeUtils.getMinusSecond(prohibitDrivingData.AlarmEndTime, prohibitDrivingData.AlarmBeginTime) / 60;
                                    int thirdDuration = severityAlarmSetting.getYjcxsj() != null ? severityAlarmSetting.getYjcxsj() : 0;
                                    int secondDuration = severityAlarmSetting.getYjcxsj2() != null ? severityAlarmSetting.getYjcxsj2() : 0;
                                    if (interval >= thirdDuration) {
                                        //三级报警
                                        if (prohibitDrivingData.AlarmLevel < 3) {
                                            prohibitDrivingData.AlarmLevel = 3;
                                            prohibitDrivingResult.setKafkaVehicleNormalAlarm(normalAlarmService.SendAlarm(prohibitDrivingData, gpsPosition, true, driverLoginData));
                                        }
                                    } else if (interval >= secondDuration) {
                                        //二级报警
                                        if (prohibitDrivingData.AlarmLevel < 2) {
                                            prohibitDrivingData.AlarmLevel = 2;
                                            prohibitDrivingResult.setKafkaVehicleNormalAlarm(normalAlarmService.SendAlarm(prohibitDrivingData, gpsPosition, true, driverLoginData));
                                        }
                                    }
                                }
                            } else {
                                //未满足条件，计算是否发送预警。夜间禁行得预警
                            }
                        } else {
                            //速度小于判定值
                            if (prohibitDrivingData.IsAlarm) {
                                //从报警状态切换未报警状态
                                prohibitDrivingData.StopAlarm(gpsPosition);
                                //从报警切换到没报警，发送停止报警
                                prohibitDrivingResult.setKafkaVehicleNormalAlarm(normalAlarmService.SendAlarm(prohibitDrivingData, gpsPosition, false, driverLoginData));

                            }
                            prohibitDrivingData.GenerateSingleGuid();
                            prohibitDrivingData.ResetData();
                        }
                    } else {
                        //不在时间范围内
                        if (prohibitDrivingData.IsAlarm) {
                            //从报警状态切换未报警状态
                            prohibitDrivingData.StopAlarm(gpsPosition);
                            //从报警切换到没报警，发送停止报警
                            prohibitDrivingResult.setKafkaVehicleNormalAlarm(normalAlarmService.SendAlarm(prohibitDrivingData, gpsPosition, false, driverLoginData));

                        }
                        prohibitDrivingData.GenerateSingleGuid();
                        prohibitDrivingData.ResetData();
                    }
                } else {
                    //关闭了规则
                    if (prohibitDrivingData.IsAlarm) {
                        //从报警状态切换未报警状态
                        prohibitDrivingData.StopAlarm(gpsPosition);
                        //从报警切换到没报警，发送停止报警
                        prohibitDrivingResult.setKafkaVehicleNormalAlarm(normalAlarmService.SendAlarm(prohibitDrivingData, gpsPosition, false, driverLoginData));

                    }
                    prohibitDrivingData.GenerateSingleGuid();
                    prohibitDrivingData.ResetData();
                }
            } else {
                //找不到规则
                if (prohibitDrivingData.IsAlarm) {
                    //从报警状态切换未报警状态
                    prohibitDrivingData.StopAlarm(gpsPosition);
                    //从报警切换到没报警，发送停止报警
                    prohibitDrivingResult.setKafkaVehicleNormalAlarm(normalAlarmService.SendAlarm(prohibitDrivingData, gpsPosition, false, driverLoginData));
                }
                prohibitDrivingData.GenerateSingleGuid();
                prohibitDrivingData.ResetData();
            }

            clusterRedisTemplate.opsForValue().set(RedisKeyPrefixConfig.RedisKey_ProhibitDriving + gpsPosition.getVehId(), prohibitDrivingData, 8, TimeUnit.HOURS);


        } catch (Exception e) {
            e.printStackTrace();
            sqlSession.rollback();
        } finally {
            sqlSession.close();
            return prohibitDrivingResult;
        }


    }


    public ProhibitDrivingData TryGetData_PDD(String vehicleId, RedisTemplate<String, Object> clusterRedisTemplate) {
        Object obj = clusterRedisTemplate.opsForValue().get(RedisKeyPrefixConfig.RedisKey_ProhibitDriving + vehicleId);
        if (obj != null) {
            return JSON.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), ProhibitDrivingData.class);
        } else {
            return new ProhibitDrivingData();
        }
    }
}
